package com.zycfc.zsf.boot.mq.rabbitmq.retry;

import com.zycfc.zsf.boot.mq.rabbitmq.producer.*;
import javax.annotation.*;
import java.util.*;
import com.zycfc.zsf.boot.util.json.*;
import java.time.*;
import com.zycfc.zsf.boot.mq.rabbitmq.util.*;
import org.slf4j.*;

public class RetryThread implements Runnable
{
    private static final Logger log;
    private static final String TIMEOUT_FILE_NAME_PREFIX = "timeout";
    private static final String UTF_8 = "UTF-8";
    private final SimpleMessageProducer messageProducer;
    private final RetryCache retryCache;
    private final RetryCacheProperties properties;
    private volatile boolean running;
    private volatile boolean paused;
    
    public RetryThread(final RetryCache retryCache, final SimpleMessageProducer messageProducer, final RetryCacheProperties properties) {
        this.running = false;
        this.paused = false;
        this.retryCache = retryCache;
        this.messageProducer = messageProducer;
        this.properties = properties;
    }
    
    @PostConstruct
    public void start() {
        final Thread thread = new Thread(this);
        this.running = true;
        thread.start();
    }
    
    public void pause() {
        this.paused = true;
    }
    
    public boolean isPaused() {
        return this.paused;
    }
    
    public void resume() {
        this.paused = false;
    }
    
    @PreDestroy
    public void stop() {
        this.running = false;
    }
    
    @Override
    public void run() {
        while (this.running) {
            if (this.isPaused()) {
                try {
                    Thread.sleep(100L);
                }
                catch (InterruptedException e) {
                    RetryThread.log.error(e.getMessage(), (Throwable)e);
                    Thread.currentThread().interrupt();
                }
            }
            else {
                this.doCheck();
                try {
                    Thread.sleep(1000L);
                }
                catch (InterruptedException e) {
                    RetryThread.log.error(e.getMessage(), (Throwable)e);
                    Thread.currentThread().interrupt();
                }
            }
        }
    }
    
    private void doCheck() {
        final MessageCache[] cacheQueue = this.getMessageCacheQueue();
        if (null == cacheQueue || cacheQueue.length == 0) {
            return;
        }
        final List<MessageContent> timeOutList = new ArrayList<MessageContent>((cacheQueue.length > 16) ? 0 : (cacheQueue.length / 2));
        final long now = System.currentTimeMillis();
        for (final MessageCache messageCache : cacheQueue) {
            if (null != messageCache) {
                if (now - messageCache.getFirstSendTime() > this.properties.getOverTime()) {
                    timeOutList.add(new MessageContent(messageCache));
                    this.retryCache.del(messageCache);
                }
                else if (now - messageCache.getLastSendTime() > this.properties.getRetryInterval()) {
                    if (!messageCache.isSended()) {
                        try {
                            messageCache.setLastSendTime(now);
                            RetryThread.log.debug("\u5c1d\u8bd5\u91cd\u53d1 id={}", (Object)messageCache.getId());
                            this.messageProducer.doSendMessage(messageCache.getExchangeName(), messageCache.getRoutingKey(), messageCache.getMessage(), messageCache.getId());
                        }
                        catch (Exception e) {
                            RetryThread.log.error(e.getMessage(), (Throwable)e);
                        }
                    }
                    else {
                        timeOutList.add(new MessageContent(messageCache));
                        this.retryCache.del(messageCache);
                    }
                }
            }
        }
        this.persistentTimeOutMessage(timeOutList);
    }
    
    private MessageCache[] getMessageCacheQueue() {
        final Map<String, MessageCache> messageLocalCache = this.retryCache.getMessageLocalCache();
        if (null == messageLocalCache || messageLocalCache.isEmpty()) {
            return null;
        }
        MessageCache[] tmpArr = new MessageCache[messageLocalCache.size()];
        tmpArr = messageLocalCache.values().toArray(tmpArr);
        return tmpArr;
    }
    
    private void persistentTimeOutMessage(final List<MessageContent> timeOutList) {
        if (null != timeOutList && !timeOutList.isEmpty()) {
            try {
                final String value = JsonUtils.objectToJson((Object)timeOutList);
                final String fileName = "timeout" + LocalDate.now() + ".cache";
                FileUtil.writeFile(this.properties.getLocalCachePath() + fileName, value.getBytes("UTF-8"), true);
            }
            catch (Exception e) {
                RetryThread.log.error(e.getMessage(), (Throwable)e);
            }
        }
    }
    
    static {
        log = LoggerFactory.getLogger((Class)RetryThread.class);
    }
}
